package org.elasticsearch.plugin.index.mapper.cnn;

import static org.elasticsearch.index.mapper.MapperBuilders.stringField;

import java.awt.TextField;
import java.io.IOException;
import java.util.List;
import java.util.Map;

import org.apache.lucene.document.BinaryDocValuesField;
import org.apache.lucene.document.Field.Store;
import org.apache.lucene.document.StringField;
import org.apache.lucene.util.BytesRef;
import org.elasticsearch.ElasticsearchCNNProcessException;
import org.elasticsearch.ElasticsearchIllegalArgumentException;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.collect.Lists;
import org.elasticsearch.common.collect.Maps;
import org.elasticsearch.common.hppc.cursors.ObjectObjectCursor;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.mapper.FieldMapperListener;
import org.elasticsearch.index.mapper.Mapper;
import org.elasticsearch.index.mapper.MapperParsingException;
import org.elasticsearch.index.mapper.MergeContext;
import org.elasticsearch.index.mapper.MergeMappingException;
import org.elasticsearch.index.mapper.ObjectMapperListener;
import org.elasticsearch.index.mapper.ParseContext;
import org.elasticsearch.threadpool.ThreadPool;

import org.elasticsearch.plugin.util.*;

public class CNNMapper implements Mapper {

	private static ESLogger logger = ESLoggerFactory.getLogger(CNNMapper.class.getName());

	public static final String CONTENT_TYPE = "cnn";
	public static final String HASH = "hash";
	public static final String FEATURE = "feature";
	public static final String BIT_SAMPLING_FILE = "/hash/LshBitSample.obj";
	public static final String LSH_HASH_FILE = "/hash/LSHashFunctions.obj";

	static {
		try {
			BitSample.readHashFunctions(CNNMapper.class.getResourceAsStream(BIT_SAMPLING_FILE));
			LocalitySensitiveHash.readHashFunctions(CNNMapper.class.getResourceAsStream(LSH_HASH_FILE));
		} catch (IOException e) {
			logger.error("Failed to initialize hash function", e);
		}
	}

	public static class Builder extends Mapper.Builder<Builder, CNNMapper> {

		private ThreadPool threadPool;

		private Map<String, Map<String, Object>> features = Maps.newHashMap();
		Map<String, Mapper> hashMappers = Maps.newHashMap();

		protected Builder(String name, ThreadPool threadPool) {
			super(name);
			// TODO Auto-generated constructor stub
			this.threadPool = threadPool;
			this.builder = this;
		}

		public Builder addFeature(String featureEnum, Map<String, Object> featureMap) {
			this.features.put(featureEnum, featureMap);
			return this;
		}

		@Override
		public CNNMapper build(BuilderContext context) {
			// TODO Auto-generated method stub
			Map<String, Mapper> featureMappers = Maps.newHashMap();
			context.path().add(name);

			// add feature and hash mappers
			for (String feature : features.keySet()) {
				Map<String, Object> featureMap = features.get(feature);
				// add cnn mapper
				/*featureMappers.put(feature,
						binaryField(feature).store(true).includeInAll(false).index(false).build(context));*/
				featureMappers.put(feature,
						stringField(feature).store(true).includeInAll(false).index(true).build(context));

				// add hash mapper if hash is required
				if (featureMap.containsKey(HASH)) {
					List<String> hashes = (List<String>) featureMap.get(HASH);
					for (String h : hashes) {
						String hashFieldName = feature + "." + HASH + "." + h;
						hashMappers.put(hashFieldName,
								stringField(hashFieldName).store(true).includeInAll(false).index(true).build(context));
					}
				}
			}
			return new CNNMapper(name, threadPool, context.indexSettings(), features, featureMappers, hashMappers);
		}
	}

	public static class TypeParser implements Mapper.TypeParser {
		private ThreadPool threadPool;

		public TypeParser(ThreadPool threadPool) {
			this.threadPool = threadPool;
		}

		@SuppressWarnings({ "unchecked" })
		@Override
		public Mapper.Builder parse(String name, Map<String, Object> node, ParserContext parserContext)
				throws MapperParsingException {
			// TODO Auto-generated method stub
			CNNMapper.Builder builder = new CNNMapper.Builder(name, threadPool);
			Map<String, Object> features = Maps.newHashMap();

			for (Map.Entry<String, Object> entry : node.entrySet()) {
				String fieldName = entry.getKey();
				Object fieldNode = entry.getValue();

				if (FEATURE.equals(fieldName)) {
					features = (Map<String, Object>) fieldNode;
				}
			}

			if (features == null || features.isEmpty()) {
				throw new ElasticsearchIllegalArgumentException("Feature not found");
			}

			// process features
			for (Map.Entry<String, Object> entry : features.entrySet()) {
				String feature = entry.getKey();
				Map<String, Object> featureMap = (Map<String, Object>) entry.getValue();

				// process hash for each feature
				if (featureMap.containsKey(HASH)) {
					Object hashVal = featureMap.get(HASH);
					List<String> hashes = Lists.newArrayList();
					if (hashVal instanceof List) {
						for (String h : (List<String>) hashVal) {
							hashes.add(HashEnum.valueOf(h).name());
						}
					} else if (hashVal instanceof String) {
						hashes.add(HashEnum.valueOf((String) hashVal).name());
					} else {
						throw new ElasticsearchIllegalArgumentException("Malformed hash value");
					}
					featureMap.put(HASH, hashes);
				}
				builder.addFeature(feature, featureMap);
			}
			return builder;
		}
	}

	private final String name;
	private final ThreadPool threadPool;
	private final Settings settings;
	private volatile ImmutableOpenMap<String, Map<String, Object>> features = ImmutableOpenMap.of();
	private volatile ImmutableOpenMap<String, Mapper> featureMappers = ImmutableOpenMap.of();
	private volatile ImmutableOpenMap<String, Mapper> hashMappers = ImmutableOpenMap.of();

	public CNNMapper(String name, ThreadPool threadPool, Settings settings, Map<String, Map<String, Object>> features,
			Map<String, Mapper> featureMappers, Map<String, Mapper> hashMappers) {
		this.name = name;
		this.threadPool = threadPool;
		this.settings = settings;
		if (features != null) {
			this.features = ImmutableOpenMap.builder(this.features).putAll(features).build();
		}
		if (featureMappers != null) {
			this.featureMappers = ImmutableOpenMap.builder(this.featureMappers).putAll(featureMappers).build();
		}
		if (hashMappers != null) {
			this.hashMappers = ImmutableOpenMap.builder(this.hashMappers).putAll(hashMappers).build();
		}
	}

	//@Override
	public String name() {
		// TODO Auto-generated method stub
		return name;
	}

	@Override
	public void parse(ParseContext context) throws IOException {
		// TODO Auto-generated method stub
		String content = null;
		XContentParser parser = context.parser();
		XContentParser.Token token = parser.currentToken();
		
		if (token == XContentParser.Token.VALUE_STRING) { 
			content = parser.text(); 
		}
		 
		/*double[] content = new double[4096];
		while (token == XContentParser.Token.VALUE_NUMBER) {
				content[i++] = parser.doubleValue();
				token = parser.nextToken();
		}*/
		if (content == null) {
			throw new MapperParsingException("No content is provided.");
		}
		double[] data = SerializationUtils.toDoubleArray(content);

		for (ObjectObjectCursor<String, Map<String, Object>> cursor : features) {
			String feature = cursor.key;
			Map<String, Object> featureMap = cursor.value;

			try {
				Mapper featureMapper = featureMappers.get(feature);
				context.externalValue(content);
				featureMapper.parse(context);
				//BytesRef r = new BytesRef(content);
				//System.out.println(name() + "." + feature + " " + content + " " + r.toString());
				context.doc().add(new BinaryDocValuesField(name() + "." + feature, new BytesRef(SerializationUtils.toByteArray(data))));
				//context.doc().add(new StringField(name() + "." + feature, content, Store.NO));
				// add hash if required
				if (featureMap.containsKey(HASH)) {
					List<String> hashes = (List<String>) featureMap.get(HASH);
					for (String h : hashes) {
						HashEnum hashEnum = HashEnum.valueOf(h);
						int[] hashVals = null;
						if (hashEnum.equals(HashEnum.BIT_SAMPLING)) {
							hashVals = BitSample.generateHashes(data);
						} else if (hashEnum.equals(HashEnum.LSH)) {
							hashVals = LocalitySensitiveHash.generateHashes(data);
						}

						String mapperName = feature + "." + HASH + "." + h;
						Mapper hashMapper = hashMappers.get(mapperName);
						context.externalValue(SerializationUtils.arrayToString(hashVals));
						hashMapper.parse(context);
					}
				}
			} catch (Exception e) {
				throw new ElasticsearchCNNProcessException("Failed to index feature " + feature, e);
			}
		}

	}

	//@Override
	public void merge(Mapper mergeWith, MergeContext mergeContext) throws MergeMappingException {
		// TODO Auto-generated method stub

	}

	//@Override
	public void traverse(FieldMapperListener fieldMapperListener) {
		for (ObjectObjectCursor<String, Mapper> cursor : featureMappers) {
			cursor.value.traverse(fieldMapperListener);
		}
		for (ObjectObjectCursor<String, Mapper> cursor : hashMappers) {
			cursor.value.traverse(fieldMapperListener);
		}
	}

	//@Override
	public void traverse(ObjectMapperListener objectMapperListener) {
		// TODO Auto-generated method stub

	}

	//@Override
	public void close() {
		// TODO Auto-generated method stub

	}

	//@Override
	public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
		builder.startObject(name);

		builder.field("type", CONTENT_TYPE);

		builder.startObject(FEATURE);
		for (ObjectObjectCursor<String, Map<String, Object>> cursor : features) {
			builder.field(cursor.key, cursor.value);
		}
		builder.endObject();

		builder.endObject();
		return builder;
	}
}
